package com.rem.flink.flink3Transform;


import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 富函数类（Rich Function Classes）
 * 已抽象类的形式出现 RichMapFunction、RichFilterFunction、RichReduceFunction
 * 与常规函数比 可以获取运行环境的上下文，并拥有一些生命周期方法，所以可以实现更复杂的功能。
 *
 * @author Rem
 * @date 2022/10/9
 */
public class TransformRichFunctionTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L),
                new Event("Alice", "./prod?id=1", 5 * 1000L),
                new Event("Cary", "./home", 60 * 1000L)
        );

        stream.map(new RichMapFunction<Event, Long>() {

            /**
             * 初始化方法，也就是会开启一个算子的生命周期
             * 当一个算子方法被调用前 open()会首先被调用
             * @param parameters
             * @throws Exception
             */
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("索引为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务开始");
            }



            @Override
            public Long map(Event event) {
                return event.getTimestamp();
            }
            /**
             * 生命周期的最后一个方法
             * 对于一个并行的自认为来说只会调用一次，例如map操作 每条数据到来后都会触发一次调用
             * @throws Exception
             */
            @Override
            public void close() throws Exception {
                System.out.println("索引为 " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务结束");
            }
        }).print();

        //filter操作
        //filter(stream);

        //map操作
        //map(stream);

        //flatmap操作
        //flatMap(stream);

        env.execute();
    }

    private static void flatMap(DataStreamSource<Event> stream) {
        stream.flatMap((event, collector) -> {
                    if ("Bob".equals(event.getUser())) {
                        collector.collect(event.getUser() + ":" + event.getUrl());
                    }
                })
                .returns(new TypeHint<>() {
                })
                .print();
    }

    private static void filter(DataStreamSource<Event> stream) {
        stream.filter(a -> "Bob".equals(a.getUser())).print();
    }

    private static void map(DataStreamSource<Event> stream) {
        stream.map(Event::getUser).print();
    }

}

